package com.camemax.controller;

/*
*   2021-06-02 17:36:01
*   订单实时对账
* */

import com.camemax.operators.OrderCompareCoProcessFunction;
import com.camemax.pojo.DealCompareEvent;
import com.camemax.pojo.OrderPayEvent;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.OutputTag;

public class OrderCompare {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        DataStream<String> orderEventStream = env.readTextFile(OrderCompare.class.getResource("/OrderLog.csv").getPath());
        DataStream<String> payEventStream = env.readTextFile(OrderCompare.class.getResource("/ReceiptLog.csv").getPath());

        // 准备数据流 —— OrderLog.csv
        DataStream<OrderPayEvent> orderLog = orderEventStream.map((MapFunction<String, OrderPayEvent>) (data -> {
            String[] fields = data.split(",");
            return new OrderPayEvent(
                    Long.parseLong(fields[0])
                    , fields[1]
                    , fields[2]
                    , Long.parseLong(fields[3])
            );
        })).assignTimestampsAndWatermarks(WatermarkStrategy.<OrderPayEvent>noWatermarks().withTimestampAssigner(new SerializableTimestampAssigner<OrderPayEvent>() {
                @Override
                public long extractTimestamp(OrderPayEvent element, long recordTimestamp) {
                    return element.getOrderTime() * 1000L;
                }
        })).filter( data -> !"".equals(data.getOrderToken()) );

        // 准备数据流 —— ReceiptLog.csv
        DataStream<DealCompareEvent> receiptLog = payEventStream.map((MapFunction<String, DealCompareEvent>) (data -> {
            String[] fields = data.split(",");
            return new DealCompareEvent(
                    fields[0]
                    , fields[1]
                    , Long.parseLong(fields[2])
            );
        })).assignTimestampsAndWatermarks(WatermarkStrategy.<DealCompareEvent>noWatermarks()
                .withTimestampAssigner(new SerializableTimestampAssigner<DealCompareEvent>() {
                    @Override
                    public long extractTimestamp(DealCompareEvent element, long recordTimestamp) {
                        return element.getDealTimestamp() * 1000L;
                    }
                }));

        SingleOutputStreamOperator<Tuple2<OrderPayEvent, DealCompareEvent>> connDataStream = orderLog.keyBy(OrderPayEvent::getOrderToken)
                .connect(receiptLog.keyBy(DealCompareEvent::getOrderToken))
                .process(new OrderCompareCoProcessFunction());

        connDataStream.print("match-order");
        connDataStream.getSideOutput(new OutputTag<OrderPayEvent>("order-timeout"){}).print("order-timeout");
        connDataStream.getSideOutput(new OutputTag<DealCompareEvent>("third-party-timeout"){}).print("third-party-timeout");

        env.execute();
    }
}
